﻿#define DDCL_CORE

#include "ddclservice.h"
#include "ddclthread.h"
#include "ddcltimer.h"
#include "ddclmalloc.h"
#include "ddclmap.h"
#include "ddclerr.h"
#include "ddcllog.h"
#include "ddclarray.h"

#include <stdlib.h>
#include <stdio.h>
#include <memory.h>
#include <assert.h>


#define DEFAULT_QUEUE_SIZE      32
#define DDCLSERVICE_WAITMS      5000
#define MSG_DATA_BUFF      0


typedef struct tag_MsgQueue{
    struct tag_Service * service;
}MsgQueue;

typedef struct tag_SvrThread{
    ddcl_Cond cond;
    ddcl_SpinLock busy;
    ddcl_Thread t;
    int worker;
}SvrThread;

typedef struct tag_Service{
    ddcl_SpinLock lock;
    dduint32 cap;
    dduint32 head;
    dduint32 tail;
    dduint32 queue_count;
    struct tag_Service * next;
    int in_global; // 是否在全局队列里面
    int worker; // 是否是工作线程调度的队列
    int exit; //是否已经退出
    ddcl_Msg * queue;
    ddcl_Msg cache_queue[DEFAULT_QUEUE_SIZE];
    ddcl_Service handle;
    void * ud;
    dduint32 session_idx;
    ddcl_MsgCB cb;
    ddcl_ExitServiceCB exit_cb;
    SvrThread * st;
}Service;

typedef struct tag_GlobalQueue{
    ddcl_SpinLock lock;
    Service * head;
    Service * tail;
}GlobalQueue;

static GlobalQueue _Q = { 0 };

static struct tag_HandleStorage{
    ddcl_Storage * hs;
    ddcl_RWLock lock;
} _H = { 0 };

static struct tag_Monitor{
    int exit;
    ddcl_SpinLock t_lock;
    ddcl_Map * threads;
} _M;

#define CK_OPENED(err) if(!ddcl_is_open(DDCL_SERVICE_MODULE)){ return err; }

static void
push_global_queue (Service * svr){
    assert(svr->next == NULL);
    ddcl_lock_spin(&(_Q.lock));
    if(_Q.tail) {
        _Q.tail->next = svr;
        _Q.tail = svr;
    } else {
        _Q.head = _Q.tail = svr;
    }
    ddcl_unlock_spin(&(_Q.lock));
}

static Service *
pop_global_queue (){
    ddcl_lock_spin(&(_Q.lock));
    Service * svr = _Q.head;
    if(svr){
        _Q.head = svr->next;
        if(!_Q.head){
            assert(svr == _Q.tail);
            _Q.tail = NULL;
        }
        svr->next = NULL;
    }
    ddcl_unlock_spin(&(_Q.lock));
    return svr;
}

static ddcl_Service
register_handle (Service ** s){
    ddcl_Service h;
    ddcl_wlock_rw(&(_H.lock));
    h = ddcl_register_in_storage(_H.hs, (void **)s);
    ddcl_wunlock_rw(&(_H.lock));
    return h;
}

static Service *
grep_handle (ddcl_Service h){
    Service * s;
    ddcl_rlock_rw(&(_H.lock));
    s = ddcl_find_in_storage(_H.hs, h);
    ddcl_runlock_rw(&(_H.lock));
    return s;
}

static void
del_handle (ddcl_Service h){
    ddcl_wlock_rw(&(_H.lock));
    ddcl_del_in_storage(_H.hs, h);
    ddcl_wunlock_rw(&(_H.lock));
}

static Service *
create_service (Service * svr, int worker) {
    ddcl_init_spin(&(svr->lock));
    svr->cap = DEFAULT_QUEUE_SIZE;
    svr->queue = ddcl_malloc(sizeof(ddcl_Msg) * svr->cap);
    svr->next = NULL;
    svr->session_idx = 0;
    svr->worker = worker;
    svr->queue_count = 0;
    svr->st = NULL;
    return svr;
}

static void
expand_service_queue (Service * svr, dduint32 new_cap){
    if (new_cap <= DEFAULT_QUEUE_SIZE) {
        return;
    }
    ddcl_Msg * new_queue = ddcl_malloc(sizeof(ddcl_Msg) * new_cap);
    for (unsigned i = 0; i < svr->cap; i++) {
        new_queue[i] = svr->queue[(svr->head + i) % svr->cap];
    }

    if (svr->cap > DEFAULT_QUEUE_SIZE) {
        ddcl_free(svr->queue);
    }
    svr->head = 0;
    svr->tail = svr->cap;
    svr->cap = new_cap;
    svr->queue = new_queue;
}

static void
push_service_queue (Service * svr, ddcl_Service self, int ptype, 
    int cmd, ddcl_Session session, const char * data, size_t sz, int free){
    ddcl_lock_spin(&(svr->lock));
    svr->queue_count++;
    ddcl_Msg * msg = &(svr->queue[svr->tail]);
    msg->ptype = ptype;
    msg->ud = svr->ud;
    msg->from = self;
    msg->self = svr->handle;
    msg->session = session;
    msg->data = (void *)data;
    msg->sz = sz;
    msg->cmd = cmd;
    msg->free = free;
    if (++svr->tail >= svr->cap) {
        svr->tail = 0;
    }
    if (svr->head == svr->tail) {
        expand_service_queue(svr, svr->cap * 2);
    }
    if (svr->worker && !svr->in_global) {
        svr->in_global = 1;
        push_global_queue(svr);
    }

    ddcl_unlock_spin(&(svr->lock));
}

static int
pop_service_queue (Service * svr, ddcl_Msg * msg, int max) {
    int ret = 0;
    ddcl_lock_spin(&(svr->lock));
    while (svr->head != svr->tail && ret < max) {
        msg[ret] = svr->queue[svr->head++];
        svr->queue_count --;
        ret += 1;
        int head = svr->head;
        int tail = svr->tail;
        int cap = svr->cap;

        if (head >= cap) {
            svr->head = head = 0;
        }
    }
    if(svr->queue_count == 0 && svr->cap > (DEFAULT_QUEUE_SIZE * 2)){
        ddcl_free(svr->queue);
        svr->cap = DEFAULT_QUEUE_SIZE;
        svr->queue = svr->cache_queue;
        svr->head = svr->tail = 0;
    }
    ddcl_unlock_spin(&svr->lock);

    return ret;
}

static void
free_service_queue (Service * svr){
    if (svr->exit_cb) {
        svr->exit_cb(svr->handle, svr->ud);
    }
    ddcl_destroy_spin(&(svr->lock));
    ddcl_Msg * msg;
    while (svr->head != svr->tail) {
        msg = &(svr->queue[svr->head++]);
        if(msg->data && msg->free){
            ddcl_free((void *)msg->data);
        }
        if (svr->head >= svr->cap) {
            svr->head = 0;
        }
    }
    ddcl_free(svr->queue);
    del_handle(svr->handle);
}

void
dispatch_msg (Service * svr, ddcl_Msg * msg){
    svr->cb(msg);
    if(msg->data && msg->free)
        ddcl_free((void *)msg->data);
}

int
dispatch_queue (){
    Service * svr = pop_global_queue();
    if (!svr){
        return 0;
    }
    ddcl_Msg msg[8];
    int ret = pop_service_queue(svr, msg, 8);
    if(!ret){
        return 0;
    }
    for (int i = 0; i < ret; i ++){
        dispatch_msg(svr, &(msg[i]));
    }
    if (svr->exit) {
        free_service_queue(svr);
    }else {
        ddcl_lock_spin(&(svr->lock));
        if (svr->tail != svr->head) {
            svr->in_global = 1;
        }else {
            svr->in_global = 0;
        }
        if (svr->in_global) {
            push_global_queue(svr);
        }
        ddcl_unlock_spin(&(svr->lock));
    }
    return 1;
}

static inline ddcl_Session
new_session (Service * svr){
    ddcl_Session id = ++ svr->session_idx;
    if(id == 0){
        svr->session_idx ++;
        return 1;
    }
    return id;
}

static void *
worker_fn (void * arg){
    SvrThread * st = (SvrThread *)arg;
    MsgQueue * q = NULL;
    int non_count = 0;
    while(!_M.exit){
        ddcl_lock_spin(&(st->busy));
        while((dispatch_queue() || dispatch_queue()) && !_M.exit){
        }
        ddcl_unlock_spin(&(st->busy));
        ddcl_wait_time_cond(&(st->cond), DDCLSERVICE_WAITMS);
    }
    return NULL;
}


DDCLAPI int
ddcl_open_service_module (){
    if(!ddcl_is_open(DDCL_TIMER_MODULE)){
        return DDCL_DEPEND_MODULE_NOT_OPENED;
    }
    ddcl * conf = ddcl_conf();
    memset(&_Q, 0, sizeof(GlobalQueue));
    ddcl_init_spin(&(_Q.lock));

    _H.hs = ddcl_new_storage(sizeof(Service), 0xFF + 1);
    ddcl_init_rw(&(_H.lock));

    _M.exit = 0;
    _M.threads = ddcl_new_map(8);
    ddcl_init_spin(&(_M.t_lock));

    for (dduint32 i = 0; i < conf->worker; i ++){
        SvrThread * st = ddcl_malloc(sizeof(SvrThread));
        memset(st, 0, sizeof(SvrThread));
        st->worker = 1;
        ddcl_init_spin(&(st->busy));
        ddcl_init_cond(&(st->cond));
        ddcl_new_thread(&(st->t), worker_fn, st, 0);
        ddcl_put_map(_M.threads, &(st->t), sizeof(ddcl_Thread), st);
    }
    return 0;
}

DDCLAPI void
ddcl_close_service_module (){
    _M.exit = 1;
    ddcl_signal_monitor();
    ddcl_begin_map(_M.threads);
    ddcl_Thread * t;
    size_t tsz;
    SvrThread * st;
    while(ddcl_next_map(_M.threads,
                (const void **)&t, &tsz, (const void **)&st)){
        if(st->worker){
            ddcl_exit_thread(*t);
            ddcl_join_thread(*t);
        }
    }

    ddcl_begin_storage(_H.hs);
    Service * svr;
    ddcl_Handle h;
    while(ddcl_next_storage(_H.hs, &h, (void **)&svr)){
        if (svr->worker) {
            free_service_queue(svr);
        }
    }
    ddcl_free_storage(_H.hs);
}

DDCLAPI int
ddcl_exit_service (ddcl_Service h){
    Service * svr = grep_handle(h);
    if (svr){
        ddcl_lock_spin(&(svr->lock));
        svr->exit = 1;
        ddcl_unlock_spin(&(svr->lock));
        ddcl_send(h, h, DDCL_PTYPE_SEND, DDCL_CMD_EXIT,0, NULL, 0);
        return 0;
    }else{
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
}

DDCLAPI int
ddcl_exit_service_cb(ddcl_Service h, ddcl_ExitServiceCB cb){
    CK_OPENED(DDCL_MODULE_NOT_OPENED);
    Service * svr = grep_handle(h);
    if (svr){
        svr->exit_cb = cb;
        return 0;
    }else{
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
}

DDCLAPI ddcl_Service
ddcl_new_service (ddcl_MsgCB cb, void * ud){
    CK_OPENED(0);
    Service * svr;
    ddcl_Service handle = register_handle(&svr);
    create_service(svr, 1);
    svr->cb = cb;
    svr->ud = ud;
    svr->handle = handle;
    return handle;
}

DDCLAPI ddcl_Service
ddcl_new_service_not_worker (ddcl_MsgCB cb, void * ud) {
    CK_OPENED(0);
    Service * svr;
    ddcl_Service handle = register_handle(&svr);
    create_service(svr, 0);
    svr->cb = cb;
    svr->ud = ud;
    svr->handle = handle;
    return handle;
}

DDCLAPI int
ddcl_send (ddcl_Service to, ddcl_Service self, int ptype,
    int cmd, ddcl_Session session, const void * data, size_t sz){
    Service * tos = grep_handle(to);
    if(!tos){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    push_service_queue(tos, self, ptype, cmd, session, data, sz, 1);
    return 0;
}

DDCLAPI int
ddcl_send_b (ddcl_Service to, ddcl_Service self, int ptype, 
    int cmd, ddcl_Session session, const void * data, size_t sz){
    Service * tos = grep_handle(to);
    if(!tos){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    char * buf = NULL;
    if (data){
        buf = ddcl_malloc(sz);
        memcpy(buf, data, sz);
    }
    push_service_queue(tos, self, ptype, cmd, session, buf, sz, 1);
    return 0;
}

DDCLAPI int
ddcl_send_raw (ddcl_Service to, ddcl_Service self, int ptype, 
    int cmd, ddcl_Session session, const void * data, size_t sz){
    Service * tos = grep_handle(to);
    if(!tos){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    push_service_queue(tos, self, ptype, cmd, session, data, sz, 0);
    return 0;
}

DDCLAPI int
ddcl_call (ddcl_Service to, ddcl_Service self, int ptype,
    int cmd, ddcl_Session * session, const void * data, size_t sz){
    Service * tos = grep_handle(to); Service * selfs = grep_handle(self);
    if (!tos || !selfs) {
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    *session = new_session (selfs);
    push_service_queue(tos, self, ptype, cmd, *session, data, sz, 1);
    return 0;
}

DDCLAPI int
ddcl_call_b (ddcl_Service to, ddcl_Service self, int ptype,
    int cmd, ddcl_Session * session, const void * data, size_t sz){
    Service * tos = grep_handle(to);
    Service * selfs = grep_handle(self);
    if(!tos || !selfs){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    *session = new_session (selfs);
    char * buf = NULL;
    if (data){
        buf = ddcl_malloc(sz);
        memcpy(buf, data, sz);
    }
    push_service_queue(tos, self, ptype, cmd, *session, buf, sz, 1);
    return 0;
}

DDCLAPI int
ddcl_call_raw (ddcl_Service to, ddcl_Service self, int ptype,
    int cmd, ddcl_Session * session, const void * data, size_t sz){
    Service * tos = grep_handle(to);
    Service * selfs = grep_handle(self);
    if (!tos || !selfs) {
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    *session = new_session (selfs);
    push_service_queue(tos, self, ptype, cmd, *session, data, sz, 0);
    return 0;
}

DDCLAPI int
ddcl_set_service_ud(ddcl_Service h, void * ud){
    Service * svr = grep_handle(h);
    if(svr){
        svr->ud = ud;
        return 0;
    }else{
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
}

DDCLAPI void *
ddcl_get_service_ud(ddcl_Service h){
    Service * svr = grep_handle(h);
    if(svr){
        return svr->ud;
    }else{
        return NULL;
    }
}

DDCLAPI int
ddcl_start (ddcl_Service h){
    CK_OPENED(DDCL_MODULE_NOT_OPENED);
    Service * svr = grep_handle(h);
    if (!svr){ return DDCL_SERVICE_UNKNOW_HANDLE; }
    if(svr->in_global){ return DDCL_SERVICE_UNKNOW_HANDLE; }

    ddcl_Thread t = ddcl_self_thread();
    void * v;
    if (ddcl_get_map(_M.threads,
                (const void *)&t, sizeof(t), (const void **)&v)){
        return DDCL_SERVICE_THREAD_IS_REGISTERED;
    }

    SvrThread * st = ddcl_malloc(sizeof(SvrThread));
    memset(st, 0, sizeof(SvrThread));
    st->t = t;
    ddcl_init_spin(&(st->busy));
    ddcl_init_cond(&(st->cond));
    svr->st = st;

    ddcl_lock_spin(&(_M.t_lock));
    ddcl_put_map(_M.threads, &(st->t), sizeof(ddcl_Thread), st);
    ddcl_unlock_spin(&(_M.t_lock));

    ddcl_Msg msg[16];
    while(!svr->exit){
        ddcl_lock_spin(&(st->busy));
        int ret = pop_service_queue(svr, msg, 16);
        while (ret && !svr->exit) {
            for (int i = 0; i < ret; i++){
                dispatch_msg(svr, &(msg[i]));
            }
            ret = pop_service_queue(svr, msg, 16);
        }
        if (ret) {
            for (int i = 0; i < ret; i++) {
                if (msg[i].data && msg[i].free) {
                    ddcl_free(msg[i].data);
                }
            }
        }
        ddcl_unlock_spin(&(st->busy));
        ddcl_wait_time_cond(&(st->cond), DDCLSERVICE_WAITMS);
    }
    free_service_queue(svr);
    return 0;
}

DDCLAPI int
ddcl_dispatch (ddcl_Service h){
    Service * svr = grep_handle(h);
    if(!svr){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    ddcl_Msg msg;
    while(pop_service_queue(svr, &msg, 1)){
        dispatch_msg(svr, &msg);
    }
    return 0;
}

DDCLAPI int
ddcl_timeout (ddcl_Service h, ddcl_Session * session, dduint32 ms){
    Service * svr = grep_handle(h);
    if(!svr){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    ddcl_Session sn = 0;
    if(session){
        sn = new_session(svr);
        *session = sn;
    }
    ddcl_add_timeout(h, sn, ms);
    return 0;
}

DDCLAPI void
ddcl_signal_monitor(){
    if (!_M.threads) { return; }
    ddcl_lock_spin(&(_M.t_lock));
    ddcl_begin_map(_M.threads);
    ddcl_unlock_spin(&(_M.t_lock));

    SvrThread * st;
    ddcl_Thread * t;
    size_t tsz;

    for(;;){
        ddcl_lock_spin(&(_M.t_lock));
        if(!ddcl_next_map(_M.threads,
                    (const void **)&t, &tsz, (const void **)&st)){
            ddcl_unlock_spin(&(_M.t_lock));
            return;
        }
        ddcl_unlock_spin(&(_M.t_lock));
        if(ddcl_try_lock_spin(&(st->busy))){
            ddcl_unlock_spin(&(st->busy));
            ddcl_signal_cond(&(st->cond));
        }
    }
}

DDCLAPI int
ddcl_new_session(ddcl_Service h, ddcl_Session * session){
    Service * svr = grep_handle(h);
    if(!svr){
        return DDCL_SERVICE_UNKNOW_HANDLE;
    }
    *session = new_session(svr);
    return 0;
}

DDCLAPI int
ddcl_is_worker (){
    if (!_M.threads) { return 0; }
    ddcl_Thread t = ddcl_self_thread();
    SvrThread * st;

    ddcl_lock_spin(&_M.t_lock);
    if(!ddcl_get_map(_M.threads,
                (const void *)&t, sizeof(t), (const void **)&st)){
        ddcl_unlock_spin(&_M.t_lock);
        return 0;
    }else{
        ddcl_unlock_spin(&_M.t_lock);
        return st->worker;
    }
    return 0;
}
